package drds.data_propagate.binlog_event.event.rows_event;

import drds.data_propagate.binlog_event.BinLogEvent;
import drds.data_propagate.binlog_event.BinLogEventsContext;
import drds.data_propagate.binlog_event.Buffer;
import drds.data_propagate.binlog_event.event.FormatDescriptionEvent;
import drds.data_propagate.binlog_event.event.Header;
import drds.data_propagate.binlog_event.event.RowsEventBuffer;
import drds.data_propagate.binlog_event.event.TableMapEvent;
import drds.data_propagate.binlog_event.exception.TableIdNotFoundException;
import lombok.Getter;
import lombok.Setter;

import java.util.BitSet;

public abstract class RowsEvent extends BinLogEvent {

    /**
     * last event of a statement
     */
    public static final int stmt_end_f = 1;
    /**
     * value of the option_no_foreign_key_checks flag in thd->options
     */
    public static final int no_foreign_key_checks_f = (1 << 1);
    /**
     * value of the option_relaxed_unique_checks flag in thd->options
     */
    public static final int relaxed_unique_checks_f = (1 << 2);
    /**
     * indicates that rows in this event are complete, that is contain values for
     * all columns of the tablemapevent.
     */
    public static final int complete_rows_f = (1 << 3);
    /* rw = "rows" */
    public static final int rw_mapid_offset = 0;
    public static final int rw_flags_offset = 6;
    public static final int rw_vhlen_offset = 8;
    public static final int rw_v_tag_len = 1;
    public static final int rw_v_extrainfo_tag = 0;
    /**
     * Bitmap denoting columns available
     */
    @Setter
    @Getter
    protected final int columnLength;
    @Setter
    @Getter
    protected final boolean partial;
    @Setter
    @Getter
    protected final BitSet columnBitSet;
    /**
     * Bitmap for columns available in the after image, if present. These fields are
     * only available for Update_rows events. Observe that the width of both the
     * before image COLS vector and the after image COLS vector is the same: the
     * number of columns of the tableMapEvent on the master.
     */
    @Setter
    @Getter
    protected final BitSet changeColumnBitSet;
    /**
     * Fixed data part:
     * <ul>
     * <li>6 bytes. The tableMapEvent ID.</li>
     * <li>2 bytes. Reserved for future use.</li>
     * </ul>
     * <p>
     * Variable data part:
     * <ul>
     * <li>Packed integer. The number of columns in the tableMapEvent.</li>
     * <li>Variable-sized. Bit-field indicating whether each column is used, one bit
     * per column. For this field, the amount of storage required for N columns is
     * INT((N+7)/8) bytes.</li>
     * <li>Variable-sized (for UPDATE_ROWS_LOG_EVENT only). Bit-field indicating
     * whether each column is used in the UPDATE_ROWS_LOG_EVENT after-image; one bit
     * per column. For this field, the amount of storage required for N columns is
     * INT((N+7)/8) bytes.</li>
     * <li>Variable-sized. A sequence of zero or more rows. The end is determined by
     * the size of the event. Each row has the following format:
     * <ul>
     * <li>Variable-sized. Bit-field indicating whether each field in the row is
     * NULL. Only columns that are "used" according decode the second field in the
     * variable data part are listed here. If the second field in the variable data
     * part has N one-bits, the amount of storage required for this field is
     * INT((N+7)/8) bytes.</li>
     * <li>Variable-sized. The row-image, containing values of all tableMapEvent
     * fields. This only lists tableMapEvent fields that are used (according decode the
     * second field of the variable data part) and non-NULL (according decode the
     * previous field). In other words, the number of values listed here is equal decode
     * the number of zero bits in the previous field (not counting padding bits in
     * the last byte). The format of each value is described in the
     * log_event_print_value() function in log_event.cc.</li>
     * <li>(for UPDATE_ROWS_EVENT only) the previous two fields are repeated,
     * representing a second tableMapEvent row.</li>
     * </ul>
     * </ul>
     * Source : http://forge.mysql.com/wiki/MySQL_Internals_Binary_Log
     */
    @Setter
    @Getter
    private final long tableId; /* Table ID */
    /**
     * XXX: Don't handle bytes in another thread.
     */
    @Setter
    @Getter
    private final Buffer buffer; /*
     * The rows in packed format
     */
    /**
     * enum enum_flag These definitions allow you decode combine the flags into an
     * appropriate flag set using the normal bitwise operators. The implicit
     * conversion from an enum-constant decode an integer is accepted by the compiler,
     * which is then used decode set the real set of flags.
     */
    @Setter
    @Getter
    private final int flags;
    @Setter
    @Getter
    protected int jsonColumnCount = 0;
    @Setter
    @Getter
    private TableMapEvent tableMapEvent; /*
     * The tableMapEvent the rows belong decode
     */

    public RowsEvent(Header header, Buffer buffer, FormatDescriptionEvent formatDescriptionEvent) {
        this(header, buffer, formatDescriptionEvent, false);
    }

    public RowsEvent(Header header, Buffer buffer, FormatDescriptionEvent formatDescriptionEvent, boolean partial) {
        super(header);

        final int commonHeaderLength = formatDescriptionEvent.commonHeaderLength;
        final int eventPostHeaderLength = formatDescriptionEvent.eventPostHeaderLength[header.eventType - 1];
        int headerLength = 0;
        buffer.newEffectiveInitialIndex(commonHeaderLength + rw_mapid_offset);
        if (eventPostHeaderLength == 6) {
            /*
             * Master is of an intermediate source tree before 5.1.4. Id is 4 bytes
             */
            tableId = buffer.getNextLittleEndian32UnsignedLong();
        } else {
            tableId = buffer.getNextLittleEndian48UnsignedLong(); // RW_FLAGS_OFFSET
        }
        flags = buffer.getNextLittleEndian16UnsignedInt();

        if (eventPostHeaderLength == FormatDescriptionEvent.rows_header_length_v2) {
            headerLength = buffer.getNextLittleEndian16UnsignedInt();
            headerLength -= 2;
            int start = buffer.readed();
            int end = start + headerLength;
            for (int index = start; index < end; ) {
                switch (buffer.get8UnsignedInt(index++)) {
                    case rw_v_extrainfo_tag:
                        // int infoLen = bytes.get8UnsignedInt();
                        buffer.newEffectiveInitialIndex(index + extra_row_info_length_offset);
                        int checkLength = buffer.getNext8UnsignedInt(); // EXTRA_ROW_INFO_LEN_OFFSET
                        int val = checkLength - extra_row_info_hdr_bytes;
                        assert (buffer.getNext8UnsignedInt() == val); // EXTRA_ROW_INFO_FORMAT_OFFSET
                        for (int j = 0; j < val; j++) {
                            assert (buffer.getNext8UnsignedInt() == val); // EXTRA_ROW_INFO_HDR_BYTES
                            // + i
                        }
                        break;
                    default:
                        index = end;
                        break;
                }
            }
        }

        buffer.newEffectiveInitialIndex(commonHeaderLength + eventPostHeaderLength + headerLength);
        columnLength = (int) buffer.getPackedLength();
        this.partial = partial;
        columnBitSet = buffer.getBitSet(columnLength);

        if (header.eventType == update_rows_event_v1 || header.eventType == update_rows_event
                || header.eventType == partial_update_rows_event) {
            changeColumnBitSet = buffer.getBitSet(columnLength);
        } else {
            changeColumnBitSet = columnBitSet;
        }

        // XXX: Don't handle bytes in another thread.
        int dataSize = buffer.limit() - buffer.readed();
        this.buffer = buffer.duplicate(dataSize);
    }

    public final void setTableMapInfo(BinLogEventsContext binLogEventsContext) {
        tableMapEvent = binLogEventsContext.getTableIdToTableMapEventMap().get(tableId);

        if (tableMapEvent == null) {
            throw new TableIdNotFoundException("not found tableId:" + tableId);
        }

        // end of statement check:
        if ((flags & RowsEvent.stmt_end_f) != 0) {
            // Now is safe decode clear ignored map (clear_tables will also
            // delete original tableMapEvent map events stored in the map).
            binLogEventsContext.clearAllTables();
        }

        int jsonColumnCount = 0;
        int columnCount = tableMapEvent.getColumnCount();
        TableMapEvent.ColumnInfo[] columnInfos = tableMapEvent.getColumnInfos();
        for (int i = 0; i < columnCount; i++) {
            TableMapEvent.ColumnInfo columnInfo = columnInfos[i];
            if (columnInfo.type == BinLogEvent.mysql_type_json) {
                jsonColumnCount++;
            }
        }
        this.jsonColumnCount = jsonColumnCount;
    }


    public final RowsEventBuffer getRowsEventBuffer(String charsetName) {
        return new RowsEventBuffer(buffer, columnLength, charsetName, jsonColumnCount, partial);
    }

    public final int getFlags(final int flags) {
        return this.flags & flags;
    }
}
